/*
 * Copyright (c) 2013-2014, University of Lugano
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the copyright holders nor the names of it
 *       contributors may be used to endorse or promote products derived from
 *       this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */


#include <iostream>
#include "message.h"
#include "../z_hglog.h"

#include <cassert>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <arpa/inet.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/listener.h>
#include <netinet/tcp.h>

#include <set>
#include <atomic>

enum manage_m_type {
    PROPOSER_CONNECT_ACCEPTOR,
    LEANER_CONNECT_ACCEPTOR
};

struct manange_message {
    enum manage_m_type xyid;
    char str[512];
};

struct subscription {
    paxos_message_type type;
    peer_cb callback;
    void *arg;
};

struct peers {
    std::atomic_int id = {0};
    std::set<struct peer *> peers;
    std::set<struct peer *> clients;
    struct evconnlistener *listener;
    struct event_base *base;
    struct evpaxos_config *config;
    int subs_count;
    struct subscription subs[32];
};


static struct timeval reconnect_timeout = {10, 0};

static int create_manage_pipe(int fd, void *proposer_or_leaner, struct event_base *base);

static struct peer *make_peer(struct peers *p, int id,
                              struct sockaddr_in *in,
                              callback_fuc *_fuc,
                              char *disc);

static void free_all_peers(std::set<struct peer *> &p, int count);

static void connect_peer(struct peer *p);

static void peers_connect(struct peers *p, int id,
                          struct sockaddr_in *addr,
                          callback_fuc *_fuc,
                          char *disc);

static void on_read(struct bufferevent *bev, void *arg);

static void on_read_m(struct bufferevent *bev, void *arg);

static void on_read_accept(struct bufferevent *bev, void *arg);

static void on_peer_event(struct bufferevent *bev, short ev, void *arg);

static void on_client_event(struct bufferevent *bev, short events, void *arg);

static void on_connection_timeout(int fd, short ev, void *arg);

static void on_timer(int fd, short ev, void *arg);

static void on_listener_error(struct evconnlistener *l, void *arg);

static void on_accept(struct evconnlistener *l,
                      evutil_socket_t fd,
                      struct sockaddr *addr,
                      int socklen,
                      void *arg);

static void on_accept_acc(struct evconnlistener *l,
                          evutil_socket_t fd,
                          struct sockaddr *addr,
                          int socklen,
                          void *arg);

static void socket_set_nodelay(int fd);


int create_manage_pipe(int fd, void *proposer_or_leaner, struct event_base *base) {
    struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE);
    bufferevent_setcb(bev, on_read_m, NULL, NULL, proposer_or_leaner);
    bufferevent_enable(bev, EV_READ);
    return 0;
}


static void
on_read_m(struct bufferevent *bev, void *arg) {
    struct evbuffer *in = bufferevent_get_input(bev);
    size_t size = evbuffer_get_length(in);
    if (size == 0) return;

    char *buffer = (char *) evbuffer_pullup(in, size);

    long long *addr_long = (long long *) buffer;
    for (int i = 0; i < size / 8; ++i) {
        struct manange_message *t = (struct manange_message *) addr_long[i];
        if (t->xyid == PROPOSER_CONNECT_ACCEPTOR) {
            struct peers *p = reinterpret_cast<peers *>(arg);
            int id = -1;
            struct sockaddr_in a;
            peers_connect(p, id, &a, nullptr, nullptr);
        } else if (t->xyid == LEANER_CONNECT_ACCEPTOR) {
            struct peers *p = reinterpret_cast<peers *>(arg);
            int id = -1;
            struct sockaddr_in a;
            peers_connect(p, id, &a, nullptr, nullptr);
        }
    }
}

struct peers *
peers_new(struct event_base *base, struct evpaxos_config *config) {
    struct peers *p = new struct peers;
    p->id = 0;
    p->subs_count = 0;
    p->listener = NULL;
    p->base = base;
    p->config = config;
    return p;
}

void
peers_free(struct peers *p) {
    free_all_peers(p->peers, 0);
    free_all_peers(p->clients, 0);
    if (p->listener != NULL)
        evconnlistener_free(p->listener);
    free(p);
}

int
peers_count(struct peers *p) {
    return p->peers.size();
}

static void
peers_connect(struct peers *p,
              int id,
              struct sockaddr_in *addr,
              callback_fuc *_fuc,
              char *disc) {
    auto peer = make_peer(p, id, addr, _fuc, disc);
    p->peers.insert(peer);

    bufferevent_setcb(peer->bev, on_read, NULL, on_peer_event, peer);
    peer->reconnect_ev = evtimer_new(p->base, on_connection_timeout, peer);
    peer->timer_ev = evtimer_new(p->base, on_timer, peer);
    connect_peer(peer);

    struct timeval tv = {10, 0};
    event_add(peer->timer_ev, &tv);
}

void
peers_connect_to_acceptors(struct peers *p,
                           callback_fuc *_fuc,
                           char *disc) {
    int i;
    for (i = 0; i < evpaxos_acceptor_count(p->config); i++) {
        struct sockaddr_in addr = evpaxos_acceptor_address(p->config, i);
        peers_connect(p, i, &addr, _fuc, disc);
    }
}

void
peers_foreach_acceptor(struct peers *p,
                       peer_iter_cb cb,
                       void *arg) {
    for (auto &node:p->peers) {
        cb(node, arg);
    }
}

void
peers_foreach_client(struct peers *p,
                     peer_iter_cb cb,
                     void *arg) {
    for (auto &node:p->clients) {
        cb(node, arg);
    }
}

struct peer *
peers_get_acceptor(struct peers *p, int id) {
    for (auto it = p->peers.begin(); it != p->peers.end(); it++) {
        if ((*it)->id == id)
            return *it;
    }

    return NULL;
}

struct bufferevent *
peer_get_buffer(struct peer *p) {
    return p->bev;
}

int
peer_get_id(struct peer *p) {
    return p->id;
}

int peer_connected(struct peer *p) {
    return p->status == BEV_EVENT_CONNECTED;
}

int
peers_listen(struct peers *p, int port) {
    struct sockaddr_in addr;
    unsigned flags = LEV_OPT_CLOSE_ON_EXEC
                     | LEV_OPT_CLOSE_ON_FREE
                     | LEV_OPT_REUSEABLE;

    /* listen on the given port at address 0.0.0.0 */
    memset(&addr, 0, sizeof(struct sockaddr_in));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(0);
    addr.sin_port = htons(port);

    p->listener = evconnlistener_new_bind(p->base, on_accept, p, flags, -1, (struct sockaddr *) &addr, sizeof(addr));
    if (p->listener == NULL) {
        paxos_log_error("Failed to bind on port %d", port);
        return 0;
    }
    evconnlistener_set_error_cb(p->listener, on_listener_error);
    paxos_log_info("Listening on port %d", port);
    return 1;
}

int
peers_listen_acc(struct peers *p, int port) {
    struct sockaddr_in addr;
    unsigned flags = LEV_OPT_CLOSE_ON_EXEC
                     | LEV_OPT_CLOSE_ON_FREE
                     | LEV_OPT_REUSEABLE;

    /* listen on the given port at address 0.0.0.0 */
    memset(&addr, 0, sizeof(struct sockaddr_in));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(0);
    addr.sin_port = htons(port);

    p->listener = evconnlistener_new_bind(p->base, on_accept_acc, p, flags, -1, (struct sockaddr *) &addr,
                                          sizeof(addr));
    if (p->listener == NULL) {
        paxos_log_error("Failed to bind on port %d", port);
        return 0;
    }
    evconnlistener_set_error_cb(p->listener, on_listener_error);
    paxos_log_info("Listening on port %d", port);
    return 1;
}


void
peers_subscribe(struct peers *p, paxos_message_type type, peer_cb cb, void *arg) {
    struct subscription *sub = &p->subs[p->subs_count];
    sub->type = type;
    sub->callback = cb;
    sub->arg = arg;
    p->subs_count++;
}

struct event_base *
peers_get_event_base(struct peers *p) {
    return p->base;
}

static void
dispatch_message(struct peer *p, paxos_message *msg) {
    for (int i = 0; p && i < p->peers->subs_count; ++i) {
        struct subscription *sub = &p->peers->subs[i];
        if (sub->type == msg->type)
            sub->callback(p, msg, sub->arg);
    }
}

static void
on_read(struct bufferevent *bev, void *arg) {
    paxos_message msg;
    struct peer *p = (struct peer *) arg;
    if (!p) return;

    int l = recv_paxos_message(bev, p, &msg);
    if (l == 0) {
        free_peer(p);
        return;
    }

    while (1) {
        unsigned int len = 0;
        memcpy(reinterpret_cast<void *>(&len), p->msg, sizeof(int));
        if (len > 4096) {
            free_peer(p);
            break;
        } else if (p && p->len - 4 >= (int) len) {
            char tmp[4096] = {0};
            assert(p != NULL);
            memcpy(tmp, p->msg + 4, len);
            int r = unpack_paxos_message(tmp, &msg);
            if (r == 0) {
                memmove(p->msg, p->msg + 4 + len, p->len - 4 - len);
                p->len = p->len - 4 - len;
                dispatch_message(p, &msg);
                paxos_message_destroy(&msg);
            } else {
                free_peer(p);
                break;
            }
        } else {
            break;
        }
    }
}


static void
on_read_accept(struct bufferevent *bev, void *arg) {
    paxos_message msg;
    struct peer *p = (struct peer *) arg;
    if (!p) return;
    recv_paxos_message_acc(bev, p, &msg);

    while (1) {
        unsigned int len = 0;
        memcpy(reinterpret_cast<void *>(&len), p->msg, sizeof(int));
        if (len > 4096) {
            free_peer(p);
            break;
        } else if (p && p->len - 4 >= (int) len) {
            char tmp[4096] = {0};
            memcpy(tmp, p->msg + 4, len);
            int r = unpack_paxos_message(tmp, &msg);
            if (r == 0) {
                memmove(p->msg, p->msg + 4 + len, p->len - 4 - len);
                p->len = p->len - 4 - len;
                dispatch_message(p, &msg);
                paxos_message_destroy(&msg);
            } else {
                free_peer(p);
                break;
            }
        } else {
            break;
        }
    }
}


static void
on_peer_event(struct bufferevent *bev, short ev, void *arg) {
    struct peer *p = (struct peer *) arg;
    if (!p) return;

    if (ev & BEV_EVENT_CONNECTED) {
        paxos_log_info("Connected to %s:%d",
                       inet_ntoa(p->addr.sin_addr), ntohs(p->addr.sin_port));
        p->status = ev;
        if (p->cb_fuc)
            p->cb_fuc->fuc(bev);
    } else if (ev & BEV_EVENT_ERROR || ev & BEV_EVENT_EOF) {
        struct event_base *base;
        int err = EVUTIL_SOCKET_ERROR();
        paxos_log_error("%s (%s:%d)", evutil_socket_error_to_string(err),
                        inet_ntoa(p->addr.sin_addr), ntohs(p->addr.sin_port));
        base = bufferevent_get_base(p->bev);
        bufferevent_free(p->bev);
        p->bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);
        bufferevent_setcb(p->bev, on_read, NULL, on_peer_event, p);
        event_add(p->reconnect_ev, &reconnect_timeout);
        p->status = ev;
    } else {
        paxos_log_error("Event %d not handled", ev);
    }
}

static void
on_client_event(struct bufferevent *bev, short ev, void *arg) {
    struct peer *p = (struct peer *) arg;
    if (!p) return;

    if (ev & BEV_EVENT_EOF || ev & BEV_EVENT_ERROR) {
        auto &clients = p->peers->clients;
        auto it = clients.find(p);
        if (it != clients.end()) {
            clients.erase(it);
        }
        free_peer(p);
    } else {
        paxos_log_error("Event %d not handled", ev);
    }
}

static void
on_timer(int fd, short ev, void *arg) {
    struct peer *p = (struct peer *) arg;
    if (!p) return;
//    T_LOGI("disc:" << p->disc);

    if (p->cb_fuc) {
        p->cb_fuc->fuc_timer(p->cb_fuc);
        p->cb_fuc->fuc_check_lease(p->cb_fuc);
    }

    struct timeval tv = {1, 0};
    event_add(p->timer_ev, &tv);
}

static void
on_connection_timeout(int fd, short ev, void *arg) {
    connect_peer((struct peer *) arg);
}

static void
on_listener_error(struct evconnlistener *l, void *arg) {
    int err = EVUTIL_SOCKET_ERROR();
    struct event_base *base = evconnlistener_get_base(l);
    paxos_log_error("Listener error %d: %s. Shutting down event loop.", err,
                    evutil_socket_error_to_string(err));
    event_base_loopexit(base, NULL);
}

static void
on_accept(struct evconnlistener *l, evutil_socket_t fd,
          struct sockaddr *addr, int socklen, void *arg) {
    struct peer *peer;
    struct peers *peers = reinterpret_cast<struct peers *>(arg);

    peer = make_peer(peers, peers->id++, (struct sockaddr_in *) addr, nullptr, nullptr);
    peers->clients.insert(peer);

    bufferevent_setfd(peer->bev, fd);
    bufferevent_setcb(peer->bev, on_read, NULL, on_client_event, peer);
    bufferevent_enable(peer->bev, EV_READ | EV_WRITE);
    socket_set_nodelay(fd);

    paxos_log_info("Accepted connection from %s:%d",
                   inet_ntoa(((struct sockaddr_in *) addr)->sin_addr),
                   ntohs(((struct sockaddr_in *) addr)->sin_port));
}


static void
on_accept_acc(struct evconnlistener *l, evutil_socket_t fd,
              struct sockaddr *addr, int socklen, void *arg) {
    struct peer *peer;
    struct peers *peers = reinterpret_cast<struct peers *>(arg);

    peer = make_peer(peers, peers->id++, (struct sockaddr_in *) addr, nullptr, nullptr);
    peers->clients.insert(peer);

    bufferevent_setfd(peer->bev, fd);
    bufferevent_setcb(peer->bev, on_read_accept, NULL, on_client_event, peer);
    bufferevent_enable(peer->bev, EV_READ | EV_WRITE);
    socket_set_nodelay(fd);

    paxos_log_info("Accepted connection from %s:%d",
                   inet_ntoa(((struct sockaddr_in *) addr)->sin_addr),
                   ntohs(((struct sockaddr_in *) addr)->sin_port));
}

static void
connect_peer(struct peer *p) {
    bufferevent_enable(p->bev, EV_READ | EV_WRITE);
    bufferevent_socket_connect(p->bev, (struct sockaddr *) &p->addr, sizeof(p->addr));

    socket_set_nodelay(bufferevent_getfd(p->bev));
    paxos_log_info("Connect to %s:%d", inet_ntoa(p->addr.sin_addr), ntohs(p->addr.sin_port));
}

static struct peer *
make_peer(struct peers *peers,
          int id,
          struct sockaddr_in *addr,
          callback_fuc *_fuc,
          char *disc) {

    struct peer *p = new struct peer;
    p->msg = new char[32 * 1024];
    p->len = 0;
    p->id = id;
    p->addr = *addr;
    p->bev = bufferevent_socket_new(peers->base, -1, BEV_OPT_CLOSE_ON_FREE);
    p->peers = peers;
    p->reconnect_ev = nullptr;
    p->timer_ev = nullptr;
    p->status = BEV_EVENT_EOF;
    p->cb_fuc = _fuc;

    if (disc != nullptr) {
        memset(p->disc, 0, sizeof(p->disc));
        snprintf(p->disc, sizeof(p->disc), "%s", disc);
    }

    return p;
}

static void
free_all_peers(std::set<struct peer *> &p, int count) {
    for (auto &node: p) {
        free_peer(node);
    }
    p.clear();
}

void
free_peer(struct peer *p) {
    auto it = p->peers->peers.find(p);
    if (it != p->peers->peers.end())
        p->peers->peers.erase(it);

    it = p->peers->clients.find(p);
    if (it != p->peers->clients.end())
        p->peers->clients.erase(it);

    if (p->reconnect_ev != NULL)
        event_free(p->reconnect_ev);
    bufferevent_free(p->bev);
    delete p;
}

static void
socket_set_nodelay(int fd) {
    int flag = paxos_config.tcp_nodelay;
    setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int));
}
